﻿using System;
using System.Data;
using System.Diagnostics;
using System.IO;
using System.Threading.Tasks;
using Go;
using GoRpc;

namespace FileService
{
    static class TransmitFile
    {
        static public async Task<bool> Send(int block, Rpc.Methods.Name writeFile, FileStream file, Rpc.Session[] sessions, shared_strand ioStrand, chan<float> progress)
        {
            long fileLength = file.Length;
            if (0 == fileLength)
            {
                progress?.post(1);
                return true;
            }
            progress?.post(0);
            generator.children sendChildren = new generator.children();
            try
            {
                bool completed = true;
                int indexCount = 0;
                long fileOffset = 0;
                for (int i = 0; i < sessions.Length; i++)
                {
                    int j = i;
                    sendChildren.go(async delegate ()
                    {
                        int nextIndex = 0;  
                        Rpc.SendDoubleBuffer dbuff = new Rpc.SendDoubleBuffer(block * 1024, ioStrand);
                        Func<int> readFunc = delegate ()
                        {
                            long offset = fileOffset;
                            int count = (int)Math.Min(dbuff.ReadBuff.Length, fileLength - offset);
                            fileOffset += count;
                            if (0 == count)
                            {
                                return 0;
                            }
                            nextIndex = indexCount++;
                            file.Seek(offset, SeekOrigin.Begin);
                            return file.Read(dbuff.ReadBuff, 0, count);
                        };
                        try
                        {
                            dbuff.Async(readFunc);
                            sessions[j].OverSend(long.MaxValue, async delegate (long _1, long _2)
                            {
                                await dbuff.Exchange();
                                if (dbuff.Error)
                                {
                                    completed = false;
                                    return;
                                }
                                var sendData = dbuff.SendData;
                                if (0 == sendData.Count)
                                {
                                    return;
                                }
                                int index = nextIndex;
                                dbuff.Async(readFunc);
                                await Rpc.Session.Send(sendData, index);
                                progress?.post((float)fileOffset / fileLength);
                            });
                            Rpc.Result result = await sessions[j][writeFile].Call(fileLength, block * 1024);
                            if (!result.Success)
                            {
                                completed = false;
                            }
                        }
                        finally
                        {
                            generator.lock_stop();
                            await dbuff.Exchange();
                            generator.unlock_stop();
                        }
                    });
                }
                await sendChildren.wait_all();
                return completed;
            }
            finally
            {
                generator.lock_stop();
                await sendChildren.stop();
                generator.unlock_stop();
            }
        }

        public class RecvResult
        {
            public long fileLength;
            public long recvLength;
        }

        static public tuple<Action, Delegate> RecvHandler(FileStream file, RecvResult recvResult, wait_group recvWg, shared_strand ioStrand, chan<float> progress)
        {
            recvResult.recvLength = 0;
            recvResult.fileLength = 0;
            progress?.post(0);
            int nextIndex = 0;
            bool cancel = false;
            shared_strand runStrand = generator.self_strand();
            Map<int, wait_group> sortDict = new Map<int, wait_group>();
            return tuple.make(new Action(delegate ()
            {
                cancel = true;
                runStrand.dispatch(delegate ()
                {
                    foreach (var item in sortDict)
                    {
                        item.value2.done();
                    }
                });
            }), Rpc.Methods.Cast(async delegate (long fileLength, int blockLen)
            {
                assert.check(runStrand.running_in_this_thread(), null);
                recvWg.add();
                try
                {
                    if (cancel)
                    {
                        return;
                    }
                    Rpc.OverRecv[] recvs = Rpc.Session.OverRecv();
                    recvResult.fileLength = fileLength;
                    wait_group writeWg = new wait_group();
                    Rpc.RecvDoubleBuffer dbuff = new Rpc.RecvDoubleBuffer(blockLen, ioStrand);
                    while (!recvs[0].Cancel)
                    {
                        int len = await recvs[0].Recv(dbuff.RecvBuff, 0, dbuff.RecvBuff.Length);
                        if (len <= 0)
                        {
                            break;
                        }
                        await dbuff.Exchange();
                        if (cancel || dbuff.Error)
                        {
                            return;
                        }
                        if (recvs[0].Index != nextIndex)
                        {
                            writeWg.add();
                            sortDict.Add(recvs[0].Index, writeWg);
                            await writeWg.wait();
                            if (cancel)
                            {
                                return;
                            }
                        }
                        dbuff.Async(delegate ()
                        {
                            file.Write(dbuff.WriteData, 0, len);
                            recvResult.recvLength += len;
                            progress?.post((float)recvResult.recvLength / recvResult.fileLength);
                        });
                        nextIndex++;
                        if (sortDict.Count > 0 && sortDict.First.Key == nextIndex)
                        {
                            wait_group nextWg = sortDict.First.Value;
                            sortDict.Remove(sortDict.First);
                            nextWg.done();
                        }
                    }
                    await dbuff.Exchange();
                }
                finally
                {
                    generator.lock_stop();
                    await generator.send_strand(ioStrand, delegate ()
                    {
                        try
                        {
                            file.Flush();
                        }
                        catch (Exception)
                        {
                            recvResult.recvLength = -1;
                        }
                    });
                    recvWg.done();
                    generator.unlock_stop();
                }
            }));
        }
    }
}
